package alfred.camel;

import alfred.mdp.worker.Worker;
import alfred.mdp.worker.WorkerService;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.DefaultConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZContext;
import org.zeromq.ZDispatcher;
import org.zeromq.ZMsg;

/**
 */
public class AlfredConsumer extends DefaultConsumer {
    private static final Logger LOG = LoggerFactory.getLogger(AlfredConsumer.class);
    private final ZContext zmqContext;
    private final ZDispatcher dispatcher;
    private final String serviceName;
    private Worker worker;

    public AlfredConsumer(AlfredEndpoint endpoint, Processor processor, ZContext zmqContext, ZDispatcher dispatcher, String serviceName) {
        super(endpoint, processor);
        this.zmqContext = zmqContext;
        this.dispatcher = dispatcher;
        this.serviceName = serviceName;
    }

    @Override
    protected void doStart() throws Exception {
        LOG.debug("Starting Alfred consumer.");
        try {
            super.doStart();
            worker = new Worker(zmqContext, dispatcher, getEndpoint().getComponent().getWorkersUri(), new WorkerService() {
                @Override
                public ZMsg process(ZMsg input) {
                    Exchange exchange = getEndpoint().createExchange();
                    try {
                        exchange.getIn().setBody(input);
                        getProcessor().process(exchange);
                        return exchange.getOut().getBody(ZMsg.class);
                    } catch (Exception e) {
                        throw new RuntimeCamelException("Error in processing the exchange!\n" + exchange);
                    }
                }

                @Override
                public String getName() {
                    return serviceName;
                }
            });

        } catch (Exception ex) {
            LOG.error("Error occurred while starting Alfred Consumer.", ex);
            throw new RuntimeCamelException(ex);

        } finally {
            LOG.debug("Alfred consumer started.");

        }
    }

    @Override
    protected void doStop() throws Exception {
        LOG.debug("Stopping Alfred consumer.");
        try {
            super.doStop();
            worker.shutdown();

        } catch (Exception ex) {
            LOG.error("Error occurred while stopping Alfred Consumer.", ex);
            throw new RuntimeCamelException(ex);

        } finally {
            LOG.debug("Alfred consumer stopped.");

        }
    }

    @Override
    public AlfredEndpoint getEndpoint() {
        return (AlfredEndpoint) super.getEndpoint();
    }
}
